/* Orbas:
 *     A open source CORBA Specification implementation from Huihoo.
 *
 * Copyright 2002-2003 Huihoo.org, Inc. All Right Reserved.
 *
 * This software is licensed under LGPL license.
 * See terms of license at gnu.org.
 *
 * For more information, visit:
 *
 * http://www.huihoo.org/orbas
 */

package org.huihoo.orbas.rtutil.threadpool;

/**
 * <p>Description: PriorityChannel</p>
 * <p>[<a href="http://gee.cs.oswego.edu/dl/classes/EDU/oswego/cs/dl/util/concurrent/intro.html">
 *  Introduction to util.concurrent package.
 * </a>]
 * <p>Copyright (c) 2003</p>
 * <p>Company: <a href="http://www.huihoo.org/orbas">huihoo.org</a></p>
 * @author <a href="http://www.huihoo.org/~cocia">Cocia Lin(cocia@163.com)</a>
 * @see <a href="http://www.huihoo.org/orbas">http://www.huihoo.org/orbas</a>
 * @version 1.0
 */
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import EDU.oswego.cs.dl.util.concurrent.Channel;

public class PriorityChannel implements Channel {
	// /////////////////////
	// Variable Section.
	private List queue = new ArrayList();
	private int maxSize;

	// //////////////////////
	// Method Section.
	/**
	 * Note:this method is not thread safe.
	 * 
	 * @param node
	 * @return
	 */
	private int getNodePos(Comparable node) {
		// compare each node in sequence.
		// this can be optimize later.
		int ret = 0;
		Iterator itr = queue.iterator();
		while (itr.hasNext()) {
			Comparable _other = (Comparable) itr.next();
			if (node.compareTo(_other) > 0) {
				break;
			}
			ret++;
		}
		return ret;
	}

	private int getCurrentPriority() {
		return 1;
	}

	private PriorityNode getInNode() {
		// get the last node in the list.
		return (PriorityNode) queue.get(queue.size() - 1);
	}

	private PriorityNode getOutNode() {
		// get the first node in the list.
		return (PriorityNode) queue.get(0);
	}

	private int getOutNodeIndex() {
		return 0;
	}

	private boolean isBufferFulled() {
		return this.getCurrentSize() >= maxSize;
	}

	private boolean isBufferEmpty() {
		return this.getCurrentSize() == 0;
	}

	public int getCurrentSize() {
		synchronized (queue) {
			return queue.size();
		}
	}

	protected PriorityNode getPriorityNode(int priority, Object value) {
		PriorityNode node = new PriorityNode(priority, value);
		return node;
	}

	public PriorityChannel(int maxSize) {
		this.maxSize = maxSize;
	}

	public void put(Object value) throws InterruptedException {
		// Wait for maxable time.
		// Long.MAX_VALUE msecs = 292471208.6775360162037037037037 years.
		long waitTime = Long.MAX_VALUE;
		offer(value, waitTime);
	}

	public boolean offer(Object value, long msecs) throws InterruptedException {
		synchronized (queue) {
			long waitMsecs = msecs;
			long beginMsecs = System.currentTimeMillis();
			// Begin to wait and check queue not full to put the value.
			while (this.isBufferFulled()) {
				if (waitMsecs <= 0) {
					return false;
				}
				try {
					queue.wait(waitMsecs);
				} catch (InterruptedException e) {
					queue.notifyAll();
					throw e;
					// return false;
				}
				//
				waitMsecs = msecs - (System.currentTimeMillis() - beginMsecs);
			} // end of wait.
			int curPriority = getCurrentPriority();
			PriorityNode node = getPriorityNode(curPriority, value);
			int index = getNodePos(node);
			queue.add(index, node);
			queue.notifyAll();
		}

		return true;
	}

	public Object take() throws InterruptedException {
		// Wait for maxable time.
		// Long.MAX_VALUE msecs = 292471208.6775360162037037037037 years.
		long waitTime = Long.MAX_VALUE;
		return poll(waitTime);
	}

	public Object poll(long msecs) throws InterruptedException {
		synchronized (queue) {
			long waitMsecs = msecs;
			long beginMsecs = System.currentTimeMillis();
			// Begin to wait and check queue not full to put the value.
			while (this.isBufferEmpty()) {
				if (waitMsecs <= 0) {
					return null;
				}
				try {
					queue.wait(waitMsecs);
				} catch (InterruptedException e) {
					queue.notifyAll();
					throw e;
					// return null;
				}
				//
				waitMsecs = msecs - (System.currentTimeMillis() - beginMsecs);
			} // end of wait.
			PriorityNode node = this.getOutNode();
			queue.remove(this.getOutNodeIndex());
			queue.notifyAll();
			return node.getValue();
		}
	}

	public Object peek() {
		synchronized (queue) {
			if (this.isBufferEmpty()) {
				return null;
			} else {
				PriorityNode node = this.getOutNode();
				return node.getValue();
			}
		}
	}

}
